package com.diorsunion.hedge.io.netty;

import com.diorsunion.hedge.io.Message;
import com.diorsunion.hedge.exception.ServerException;
import com.diorsunion.hedge.io.ServerMessageHandler;
import com.diorsunion.hedge.util.LogUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.DefaultEventExecutor;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;


@Component
@Qualifier("NettyServerHandler")
@Sharable
public class NettyServerHandler extends SimpleChannelInboundHandler<Message> {
	@Resource
	ServerMessageHandler serverMessageHandler;
	ChannelGroup channelGroup = new DefaultChannelGroup(new DefaultEventExecutor());

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.info("Channel is active:" + ctx.channel().id());
		Channel channel = ctx.channel();
		channelGroup.add(channel);
		super.channelActive(ctx);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.info("Channel is disconnected:" + ctx.channel().id());
		channelGroup.remove(ctx.channel());
		super.channelInactive(ctx);
	}

	@Override
	protected void messageReceived(ChannelHandlerContext channelHandlerContext, Message msg) throws Exception {
		Channel channel = channelHandlerContext.channel();
		boolean b = channelGroup.contains(channel);
		LogUtil.debug(b ? "channel exist" + channel.id() : "channel not exist" + channel.id());
		channel.writeAndFlush(Message.genHeartBeatMessage());
		serverMessageHandler.execute(msg, x -> channel.writeAndFlush(x));
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
		LogUtil.error("netty exception", cause);
		ctx.close();
	}

	public boolean write(Message message){
		if(!channelGroup.isEmpty()){
			channelGroup.writeAndFlush(message);
			LogUtil.debug("write message success:" + message);
			return true;
		}else{
			String errormessage = "write message fail,netty not started:" + message;
			LogUtil.error(errormessage, new ServerException(errormessage));
			return false;
		}
	}

	public boolean write(Message message,String id){
		try {
			channelGroup.stream()
					.filter( x-> x!=null && x.isOpen() && x.isActive() && x.isWritable() && id.equals(x.id().asShortText()))
					.forEach( x-> write(message,x));
			return true;
		}catch (Throwable e){
			LogUtil.error("write message error:id=" + id, e);
			return false;
		}
	}

	private void write(Message message,Channel channel){
		channel.writeAndFlush(message);
		LogUtil.debug("write success:" + message);
	}

	public void closeAllChannel(){
		this.channelGroup.stream().forEach(x->x.close());
	}
}
